package com.bw.day0629;

import com.bw.gmall.realtime.app.func.LaterViewFuntion;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwsTrafficSourceKeywordPageViewWindow01 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String topic = "dws_traffic_page_log";
        String groupId = "dws_traffic_source";
        tableEnv.executeSql("" +
                " create table page_log(" +
                " `page` map<string,string>, " +
                " `ts` bigint, " +
                " `rt` as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)), " +
                " WATERMARK FOR rt AS rt - INTERVAL '2' SECOND " +
                " ) WITH ( " +
                " 'connector' = 'kafka', " +
                " 'topic' = 'dwd_traffic_page_log', " +
                "  'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                "  'properties.group.id' = 'testGroupxx',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json')");

        Table filterKeyWord = tableEnv.sqlQuery(" select " +
                " page['item'] item, " +
                " ts," +
                " rt " +
                " from page_log" +
                " where page['last_page_id'] = 'search' and " +
                " page['item_type'] = 'keyword' and page['item'] is not null ");
        tableEnv.createTemporaryView("filter_keyword", filterKeyWord);

        //炸裂
        // 注册函数
        tableEnv.createTemporaryFunction("laterviewkeyword", LaterViewFuntion.class);

        Table laterViewTable = tableEnv.sqlQuery(" select 'search' sources  ,item,word,rt from filter_keyword ," +
                "   lateral table(laterviewkeyword(item))");
        tableEnv.createTemporaryView("laterViewTable", laterViewTable);

        //10秒窗口聚合
        TableResult windowWordTable = tableEnv.executeSql("select window_start stt," +
                " window_end edt, " +
                " sources, " +
                " word, " +
                " count(*) keyword_count," +
                " UNIX_TIMESTAMP()*1000 as ts from table(\n" +
                "tumble(TABLE laterViewTable,DESCRIPTOR(rt),INTERVAL '10' SECOND)\n" +
                ")  group by  window_start,window_end,sources,word");
//                .print();
//        tableEnv.createTemporaryView("windowWordTable",laterViewTable);


        //
    }
}